/**  
  -- The Performance Driven Lime Project -- 

  A self adaptive tuple space for mobile agents and hosts by dynamic
  adapting tuple's replication policy.
  
  Copyright (C) 2008
  Zhang Jinxue, Xing Jiankuan
  SAGroup of Tsinghua University

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */

/* This is the main thread for testing, which parse the conf.xml,
 * 1. out the specific tuples
 * 2. set the specific policy
 * 3. Launch specific agents by specific parameters
 * */

package sagroup.dracots.testing;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.logging.Logger;

import lights.adapters.Field;
import lights.adapters.Tuple;
import lights.dracots.TupleSpace;
import lights.interfaces.ITuple;
import lime2.AgentCreationException;
import lime2.LimeServer;
import lime2.PolicyException;
import lime2.PropertyKeys;
import lime2.TupleID;

import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.HierarchicalConfiguration;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.commons.math.random.RandomData;
import org.apache.commons.math.random.RandomDataImpl;

import sagroup.dracots.DracoTS;
import sagroup.dracots.FlatShareTuple;
import sagroup.dracots.LogManager;
import sagroup.dracots.TupleType;
import sagroup.dracots.policy.RPTuple;
import sagroup.dracots.policy.ReplicationPolicy;

/**
 * @author Zhang Jinxue
 *
 */
public class AgentServer{

	private static final String LOG_ROOT = "Y:/DracoTS-exp/"; //log file's root directory
	static String port;
	static String xmlPath = "conf\\";
	static DracoTS dts = null;
	static Random random = new Random();
	static Logger LOG = Logger.getLogger(AgentServer.class.getName());
	static ArrayList<AgentInfo> agentInfos = new ArrayList<AgentInfo>();
	static int agentSeq = 0;
	static int liveAgentNum;
	static boolean switchOn = false;
	
	// Make a new single tuple element
	static FlatShareTuple makeNewTuple(String typeName, byte[] data){
		  ITuple it = new Tuple();
		  it.add(new Field().setValue(data));
		  return new FlatShareTuple(new TupleType(typeName),it);
	}
	
	// Set the policy for each type of tuple
	static void applyPolicy(String type,String policy){
		  
		  byte policyValue = 0;
		  
		  if (policy.compareTo("SO") == 0){
		  	  policyValue = 2;
		  } else if (policy.compareTo("RC") == 0){
		  	  policyValue = 3;
		  } else if (policy.compareTo("FR") == 0){
		  	  policyValue = 1;
		  } else if (policy.compareTo("CO") == 0){
		  	  policyValue = 4;
		  }
		  else {
			  System.out.println("Your policy is not supported now!");
			  System.exit(0);
		  }
		  
		  ReplicationPolicy rPolicy = new ReplicationPolicy(policyValue); 
		  System.out.println("Apply " +rPolicy+ " policy for type "+ type);
		  if(dts!=null){
			  RPTuple policy1 = new RPTuple(new TupleType(type), rPolicy);
			  dts.out(policy1);
		  }else{
			  System.out.println("tuple space is null!");
			  System.exit(0);
		  }
	}
	
	/**
	 * @param args
	 * @author Zhang Jinxue
	 * @throws UnknownHostException 
	 * @throws PolicyException 
	 * @throws AgentCreationException 
	 */
	
	public static void main(String[] args) throws UnknownHostException {
		
		// Obtain the port to bind
		if (args.length != 4 || !args[0].equals("-p") || !args[2].equals("-f")) {
			System.out
					.println("Usage:\tAgentServer -p PORT -f CONFIG_FILE.\n"
							+ "\t-p PORT: \tspecify the port to band\n"
							+ "\t-f CONFIG_FILE: specify the config XML file\n"
							+ "\t   \t\tThe default path is src\\sagroup\\testing\\config\\");
			System.exit(0);
		} else {
			port = args[1];
			xmlPath += args[3];
		}
		
		try {
			XMLConfiguration conf = new XMLConfiguration(xmlPath);
			
			// start the server first
			LimeServer.settings.setProperty(PropertyKeys.UCASTPORT_KEY, port);
			LimeServer server = LimeServer.getServer();
			
			switchOn = conf.getBoolean("server[@switchon]", false); // whether to turn on policy switch policy
//			PolicyDecisionMaker.setSwtichOn(on);
			
			long expTime = conf.getLong("server[@exptime]", 5) * 1000 * 60; // get the experiment time
			long sleepTime = conf.getLong("server[@sleeptime]", 2) * 1000 * 60; // get the experiment time
			AgentUtil.sleepTime = sleepTime;
			AgentUtil.expTime = expTime + AgentUtil.sleepTime;
			
			String hostName = InetAddress.getLocalHost().getHostName();
			
			//load agents
			List agentConfs = conf.configurationsAt("agents.agent");
			StringBuffer asb = new StringBuffer();
			if(!agentConfs.isEmpty()) {
				Iterator it = agentConfs.iterator();		
				for(;it.hasNext();) {
					Configuration agentConf = (Configuration)it.next();
					String className = agentConf.getString("[@classname]");
					
					Map<String, String> params = new HashMap();
					Iterator itKeys = agentConf.getKeys();
					StringBuffer asb2 = new StringBuffer();
					for(;itKeys.hasNext();) {
						String key = (String)itKeys.next();
						if(key.startsWith("section"))
							continue; //the section interval properties
						String value = agentConf.getString(key);
						key = key.substring(2, key.length() - 1); //change [@key] to key
						params.put(key, value);
						asb2.append(key);
						asb2.append("=");
						asb2.append(value);
						asb2.append(",");		
					}
					
					List sections = ((HierarchicalConfiguration)agentConf).configurationsAt("section");
					int num = (sections !=null)? sections.size() : 0;
					asb2.append("section=" + num);
					Long[][] array = new Long[num][3];
					if(sections != null)
					{
						createArray(sections,array);
					}
					AgentInfo info = new AgentInfo(className, params, array);
					agentInfos.add(info);
					asb.append("AGENTCONF,");
					asb.append(info.agentName);
					asb.append(",");
					asb.append(asb2.toString());
				}
			}
			
			//initialize log
			Date date = new Date();
			SimpleDateFormat f = new SimpleDateFormat("yyyyMMdd_HH");
			
			LogManager.init(LOG_ROOT + f.format(date) + "/", makeLogName(hostName, port, agentInfos));
			if(!agentConfs.isEmpty()) 
				LOG.info(asb.toString()); //log agent info
			
			server.boot();
			dts = DracoTS.get("testing");
			if(switchOn)
				dts.startPolicyDecisionMaker();
			String msg = String.format("SERVERCONF,%s,%s,adaptation=%b,exptime=%dms",
					hostName,
					port,
					switchOn,
					expTime);
			LOG.info(msg);
			
			System.out.println("Sleep for 60s to engage, please open all machines in the experiment...");
			Thread.sleep(60000);
			
			//load tuple size distribution
			List tupleConfs = conf.configurationsAt("tuples.tuple");
			StringBuilder tsb = new StringBuilder();
			tsb.append("INIT_TUPLES,");
			if (tupleConfs.isEmpty()) {
				tsb.append("None");
			} else {
				double mu = conf.getDouble("tuples[@mu]", 1000);
				double sigma = conf.getDouble("tuples[@sigma]", 100);
				tsb.append("mu=" + mu + ",sigma=" + sigma);
				RandomData randomData = new RandomDataImpl(); 
				
				Iterator it = tupleConfs.iterator();
				for(; it.hasNext();) {
					Configuration tupleConf = (Configuration)it.next();
					HashSet<TupleID> hashSet = new HashSet();
					String type = tupleConf.getString("[@type]");
					int num = tupleConf.getInt("[@num]");
					for (int i = 0; i < num; i++){
						int tupleSize = (int)randomData.nextGaussian(mu, sigma);
						byte[] data = new byte[tupleSize];
						random.nextBytes(data);
						FlatShareTuple tempTuple = makeNewTuple(type, data);
						hashSet.add(tempTuple.getVID());
						dts.out(tempTuple);
					}
					dts.out(ListTuple.createLocalList(new TupleType(type), hashSet));
				}
			}
			LOG.info(tsb.toString());
			
			//load policies
			List policyConfs = conf.configurationsAt("policies.policy");
			if(!policyConfs.isEmpty()) {
				StringBuilder sb = new StringBuilder();
				sb.append("INIT_POLICIES,");
				Iterator it = policyConfs.iterator();
				for(;it.hasNext();) {
					Configuration policyConf = (Configuration)it.next();
					String type = policyConf.getString("[@type]");
					String policy = policyConf.getString("[@policy]");
					sb.append(type + "=" + policy + ",");
					applyPolicy(type, policy);
				}
				LOG.info(sb.toString());
			}
			
			Thread.sleep(1000); //sleep for a while
			
			//start agents
			try {
				liveAgentNum = agentInfos.size();
				for(AgentInfo agentInfo: agentInfos) {
					
					server.loadAgent("sagroup.dracots.testing." + agentInfo.className, new Object[]{agentInfo.params,
							agentInfo.array});
				}
			} catch (AgentCreationException e) {
				LOG.severe("Error during agent creation: " + e);
				System.exit(0);
			}
			
		} catch (ConfigurationException e) {
			e.printStackTrace();
			System.exit(0);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
//	private static void createArray(List sections, Long[][] array) {
//		
//		TreeMap<Long,Long> map = new TreeMap<Long, Long>();
//		Iterator sit = sections.iterator();
//		for(;sit.hasNext();)
//		{
//			Configuration section = (Configuration)sit.next();
//			Long n = Long.valueOf(section.getString("[@n]"));
//			Long interval = Long.valueOf(section.getString("[@interval]"));
//		    map.put(n,interval);
//		}
//		int index = 0;
//		for(Long i : map.keySet())
//		{
//			array[index][0] = i;
//			array[index][1] = map.get(i);
//			index++;
//		}
//	}

	private static void createArray(List sections, Long[][] array) {
		
		int index = 0;
		
		// sort the sections by start time
		sectionsComparator comp = new sectionsComparator();
		Collections.sort(sections, comp);
		
		Iterator sit = sections.iterator();
		for(;sit.hasNext();)
		{
			Configuration section = (Configuration)sit.next();
			array[index][0] = Long.valueOf(section.getString("[@n]"));
			array[index][1] = Long.valueOf(section.getString("[@interval]"));
			if(section.getString("[@prob]") != null)
				array[index][2] = Long.valueOf(section.getString("[@prob]"));
		    index ++;
		}
	}
	
	public synchronized static void done() {
		liveAgentNum--;
		if(liveAgentNum == 0) {
			LimeServer.getServer().shutdown(false);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				//do nothing
			} // sleep a while to wait all the operations in the queue are done
			long ms = -1;
			long rs = -1;
			try{
				ms = TupleSpace.dump("testing" + port);
				rs = TupleSpace.dump("testing_replicas_" + port);
			} catch (Exception e) {
				//do nothing
			} finally {
				LOG.info(String.format("ALL_DONE,ms=%dbytes,rs=%dbytes", ms, rs));
				System.exit(0);
			}
		}	
	}

	/**
	 * Create log file name in the format <i>hostname_port_agentlist</i>
	 */
	private static String makeLogName(String hostName, String port,
		ArrayList<AgentInfo> agentInfos) {
		StringBuffer sb = new StringBuffer();
		sb.append("dracots_");
		sb.append(hostName);
		sb.append("_");
		sb.append(port);
		if (agentInfos.size() == 0)
			return sb.toString();

		sb.append("_");
		for (int i = 0; i < agentInfos.size(); i++) {
			sb.append(agentInfos.get(i).shortName);
			if (i < agentInfos.size() - 1) {
				sb.append("_");
			}
		}

		return sb.toString();
	}
	
	public static int nextAgentSeq() {
		agentSeq++;
		return agentSeq;
	}
}

class AgentInfo {
	String className;
	String shortName;
	String agentName;
	Map<String, String> params;
	Long[][] array;
	
	public AgentInfo(String className, Map<String, String> params, Long[][] array) {
		this.className = className;
		this.params = params;
		this.array = array;
		int index = className.indexOf("Agent");
		shortName = className.substring(0, index);
		agentName = shortName + AgentServer.nextAgentSeq();
		params.put("agentname", agentName);
	}
}

class sectionsComparator implements Comparator {
	public int compare(Object o1, Object o2) {        
		Configuration c1 = (Configuration) o1;     
		Configuration c2 = (Configuration) o2;     
		if (Long.valueOf(c1.getString("[@n]")) > Long.valueOf(c2.getString("[@n]")))      
			return 1;     
		else if (Long.valueOf(c1.getString("[@n]")) < Long.valueOf(c2.getString("[@n]")))        
			return 0;      
		else       
			return -1;      
	}
}